package com.atguigu.edu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.app.func.DimAsyncFunction;
import com.atguigu.edu.bean.TradeSourceOrderBeean;
import com.atguigu.edu.util.ClickHouseUtil;
import com.atguigu.edu.util.DateFormatUtil;
import com.atguigu.edu.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.time.ZoneId;
import java.util.concurrent.TimeUnit;

/**
 * @author zzw
 * @create 2022/9/5 20:04
 */
public class DwsTradeOrderSourceWindow {
    public static void main(String[] args) throws Exception {
        // TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // TODO 2 设置状态后端
        /*
        env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE );
        env.getCheckpointConfig().setCheckpointTimeout( 3 * 60 * 1000L );
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        System.setProperty("HADOOP_USER_NAME", "atguigu");
        env.setStateBackend(new HashMapStateBackend());
        */

        // 设置时区的方法
        tableEnv.getConfig().setLocalTimeZone(ZoneId.of("GMT+8"));

        // TODO 3　读取DWD层dwd_trade_order_detail 和dwd_traffic_unique_visitor_detail主题数据
        String topicName = "dwd_trade_order_pre_process";
        String uvTopic = "dwd_traffic_unique_visitor_detail";
        String groupId = "dws_trade_order_source_window";
        DataStreamSource<String> orderSourceStream = env.addSource(KafkaUtil.getKafkaConsumer(topicName, groupId));
        DataStreamSource<String> uvDS = env.addSource(KafkaUtil.getKafkaConsumer(uvTopic, groupId));

        KeyedStream<JSONObject, String> keyedStream = orderSourceStream.map(JSON::parseObject).keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject value) throws Exception {
                return value.getString("id");
            }
        });

        // TODO 5 转换数据类型 + 过滤不完整的数据
        SingleOutputStreamOperator<TradeSourceOrderBeean> jsonObjStream = keyedStream.flatMap(new FlatMapFunction<JSONObject, TradeSourceOrderBeean>() {
            @Override
            public void flatMap(JSONObject value, Collector<TradeSourceOrderBeean> out) throws Exception {
                String userId = value.getString("user_id");
                String type = value.getString("type");
                String sc = value.getString("sc");
                Long ts = value.getLong("ts");
                try {
                    if (userId != null && "insert".equals(type)) {
                        out.collect(TradeSourceOrderBeean.builder()
                                .stt("")
                                .edt("")
                                .scName("")
                                .scId(sc)
                                .sessionId(value.getString("session_id"))
                                .userId(value.getString("user_id"))
                                .uvCt(0L)
                                .orderAmount(value.getDouble("final_amount"))
                                .OrderCount(1L)
                                .ts(ts)
                                .build());
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });

        // TODO 6 转换uvDS数据类型
        SingleOutputStreamOperator<TradeSourceOrderBeean> uvBeanStream = uvDS.map(new MapFunction<String, TradeSourceOrderBeean>() {
            @Override
            public TradeSourceOrderBeean map(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                JSONObject common = jsonObject.getJSONObject("common");
                Long ts = jsonObject.getLong("ts");
                return TradeSourceOrderBeean.builder()
                        .stt("")
                        .edt("")
                        .scName("")
                        .scId(common.getString("sc"))
                        .sessionId(common.getString("sid"))
                        .userId(common.getString("uid"))
                        .orderAmount(0D)
                        .OrderCount(0L)
                        .uvCt(1L)
                        .ts(ts)
                        .build();
            }
        });

        // TODO 7 合并两个流
        DataStream<TradeSourceOrderBeean> unionBeanStream = jsonObjStream.union(uvBeanStream);

        // TODO 8 加水位线   开窗聚合
        //unionBeanStream.print("union>>>>>");

        SingleOutputStreamOperator<TradeSourceOrderBeean> withWaterMarkStream = unionBeanStream.assignTimestampsAndWatermarks(WatermarkStrategy.<TradeSourceOrderBeean>forBoundedOutOfOrderness(Duration.ofSeconds(2L)).withTimestampAssigner(new SerializableTimestampAssigner<TradeSourceOrderBeean>() {
            @Override
            public long extractTimestamp(TradeSourceOrderBeean element, long recordTimestamp) {
                return element.getTs();
            }
        }));

        SingleOutputStreamOperator<TradeSourceOrderBeean> reduceBeanStream = withWaterMarkStream.keyBy(new KeySelector<TradeSourceOrderBeean, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> getKey(TradeSourceOrderBeean value) throws Exception {
                return new Tuple2<>(value.getScId(), value.getSessionId());
            }
        }).window(TumblingEventTimeWindows.of(Time.seconds(10L))).reduce(new ReduceFunction<TradeSourceOrderBeean>() {
            @Override
            public TradeSourceOrderBeean reduce(TradeSourceOrderBeean value1, TradeSourceOrderBeean value2) throws Exception {
                value1.setOrderCount(value1.getOrderCount() + value2.getOrderCount());
                value1.setOrderAmount(value1.getOrderAmount() + value2.getOrderAmount());
                value1.setUvCt(value1.getUvCt() + value2.getUvCt());
                return value1;
            }
        }, new WindowFunction<TradeSourceOrderBeean, TradeSourceOrderBeean, Tuple2<String, String>, TimeWindow>() {
            @Override
            public void apply(Tuple2<String, String> stringStringTuple2, TimeWindow window, Iterable<TradeSourceOrderBeean> input, Collector<TradeSourceOrderBeean> out) throws Exception {
                TradeSourceOrderBeean tradeSourceOrderBeean = input.iterator().next();
                tradeSourceOrderBeean.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                tradeSourceOrderBeean.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                tradeSourceOrderBeean.setTs(System.currentTimeMillis());
                out.collect(tradeSourceOrderBeean);

            }
        });

        //reduceBeanStream.print("reduce>>>>");
        //TODO 9 关联维度数据
        SingleOutputStreamOperator<TradeSourceOrderBeean> resultBeanStream = AsyncDataStream.unorderedWait(reduceBeanStream, new DimAsyncFunction<TradeSourceOrderBeean>("DIM_BASE_SOURCE") {
            @Override
            public String getKey(TradeSourceOrderBeean input) {
                return input.getScId();
            }

            @Override
            public void join(TradeSourceOrderBeean input, JSONObject obj) {
                input.setScName(obj.getString("sourceSite"));
            }
        }, 100, TimeUnit.SECONDS);

        //TODO 10 写出到clickhouse

        resultBeanStream.print("写入ClickHouse>>>>>");

        resultBeanStream.addSink(ClickHouseUtil.getClickHouseSinkFunc("insert into dws_trade_order_source_window values(?,?,?,?,?,?,?,?,?,?)"));

        // TODO 11 执行
        env.execute(groupId);

    }
}
